package org.sunyaxing.transflow.transflowapp.plugins.outs;

import cn.hutool.core.io.FileUtil;
import com.alibaba.fastjson2.JSONObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sunyaxing.transflow.api.TransData;
import org.sunyaxing.transflow.api.common.Handle;
import org.sunyaxing.transflow.api.common.ano.Extension;
import org.sunyaxing.transflow.api.common.ano.HandleItem;
import org.sunyaxing.transflow.api.common.ano.JobParamItem;
import org.sunyaxing.transflow.api.common.ano.ScopeContentCheck;
import org.sunyaxing.transflow.api.extensions.base.ExtensionContext;
import org.sunyaxing.transflow.api.extensions.base.typesimpl.TransFlowOutputWithHandler;
import org.sunyaxing.transflow.transflowapp.controllers.JobLogsWs;
import org.sunyaxing.transflow.transflowapp.plugins.inputs.nettyfile.FileStartInfo;
import org.sunyaxing.transflow.transflowapp.plugins.inputs.nettyfile.FileUploadInfo;
import org.sunyaxing.transflow.transflowapp.plugins.inputs.nettyfile.LastFileUploadInfo;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.List;
import java.util.function.Function;

@Extension("plugin-netty-file-output")
@ScopeContentCheck(value = {
        @JobParamItem(field = "host", label = "HOST", defaultValue = "0.0.0.0"),
        @JobParamItem(field = "port", label = "端口", defaultValue = "8080"),
        @JobParamItem(field = "size", label = "分片", defaultValue = "102400"),
}, handle = @HandleItem(field = "fileLocation", label = "文件位置字段"))
public class NettyFileOutputExt extends TransFlowOutputWithHandler<String, String> {
    private static final Logger log = LoggerFactory.getLogger(NettyFileOutputExt.class);

    private String host;
    private Integer port;
    private Integer size;

    public NettyFileOutputExt(ExtensionContext extensionContext) {
        super(extensionContext);
    }

    @Override
    protected void batchExec(List<String> dataList) {

    }

    @Override
    protected void afterInitHandler(JSONObject config, List<Handle> handles) {
        this.host = config.getString("host");
        this.port = config.getInteger("port");
        this.size = config.getInteger("size");
    }

    @Override
    public Function<TransData<String>, String> parseHandleToConsumer(String handleId, String filePathKey) {
        return transData -> {
            JSONObject transJson = JSONObject.parse(transData.getData());
            String filePath = transJson.getString(filePathKey);
            JobLogsWs.sendMessage(jobId, "【{}】发送文件 {}", extensionContext.getNodeName(), filePath);
            File file = new File(filePath);

            FileStartInfo fileStartInfo = new FileStartInfo();
            fileStartInfo.setFileName(file.getName());
            fileStartInfo.setFileParentPath(file.getParent());
            fileStartInfo.setFileSize(file.length());
            fileStartInfo.setStartPos(0L);
            fileStartInfo.setPreSize(size);

            FileUploadClient fileUploadClient = new FileUploadClient(host, port);
            fileUploadClient.upload(fileStartInfo);
            return transData.getData();
        };
    }

    public static class FileUploadClient {

        private final String host;
        private final Integer port;
        private ChannelFuture future;
        private EventLoopGroup worker;

        public FileUploadClient(String host, Integer port) {
            this.host = host;
            this.port = port;
        }

        public void upload(FileStartInfo fileStartInfo) {
            worker = Epoll.isAvailable() ? new EpollEventLoopGroup() : new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            try {
                bootstrap.group(worker)
                        .channel(Epoll.isAvailable() ? EpollSocketChannel.class : NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<Channel>() {
                            @Override
                            protected void initChannel(Channel ch) throws Exception {
                                ch.pipeline().addLast(new ObjectEncoder());
                                ch.pipeline().addLast(new ObjectDecoder(ClassResolvers.weakCachingConcurrentResolver(null)));
                                ch.pipeline().addLast(new FileUploadClientHandler(fileStartInfo));
                            }
                        });
                future = bootstrap.connect(host, port).sync();
                future.channel().closeFuture().addListener((ChannelFutureListener) channelFuture -> {
                    channelFuture.channel().close();
                    worker.shutdownGracefully();
                });
            } catch (Exception e) {
                log.error("e: ", e);
                worker.shutdownGracefully();
            } finally {
            }
        }
    }


    public static class FileUploadClientHandler extends ChannelInboundHandlerAdapter {

        private final FileStartInfo fileStartInfo;
        private final RandomAccessFile accessFile;

        public FileUploadClientHandler(FileStartInfo fileStartInfo) throws IOException {
            this.fileStartInfo = fileStartInfo;
            // 定位到文件
            File file = FileUtil.file(fileStartInfo.getFileParentPath(), fileStartInfo.getFileName());
            // 获取文件通道
            this.accessFile = new RandomAccessFile(file, "r");
            // 定位读取位置
            this.accessFile.seek(fileStartInfo.getStartPos());
        }

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            // 连接激活时发送文件信息
            ctx.writeAndFlush(fileStartInfo);
        }

        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            // 接收端接到文件写入位置
            if (msg instanceof Long endPos) {
                log.info("【发送端】接收到文件 {} 已经传输到 {}", fileStartInfo.getFileName(), endPos);
                accessFile.seek(endPos);
                // 开始写入文件信息
                byte[] bytes = new byte[fileStartInfo.getPreSize()];
                int byteRead = 0;
                if ((byteRead = accessFile.read(bytes)) != -1) {
                    log.info("【发送端】封装文件数据 {}", fileStartInfo.getFileName());
                    FileUploadInfo fileUploadInfo = new FileUploadInfo();
                    // 标记接收端文件开始写入的位置
                    fileUploadInfo.setStartPos(endPos);
                    long loadPosition = endPos + byteRead;
                    log.info("【发送端】数据已经读取到 {} 当前进度 {}%", loadPosition, loadPosition * 100 / fileStartInfo.getFileSize());
                    fileUploadInfo.setByteSize(byteRead);
                    fileUploadInfo.setEndPos(loadPosition);
                    fileUploadInfo.setBytes(bytes);
                    ctx.writeAndFlush(fileUploadInfo);
                } else {
                    log.info("【发送端】封装结束数据 {}", fileStartInfo.getFileName());
                    LastFileUploadInfo lastFileUploadInfo = new LastFileUploadInfo();
                    ctx.writeAndFlush(lastFileUploadInfo);
                    accessFile.close();
                }
            }
        }

        @Override  //在当前ChannelHandler回调方法出现异常时被回调
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            log.error("发送端异常", cause);
            try {
                ctx.close();
            } catch (Exception e) {
                log.error("关闭通道失败", e);
            }
            try {
                if (this.accessFile != null) {
                    this.accessFile.close();
                }
            } catch (Exception e) {
                log.error("关闭文件失败", e);
            }
        }
    }
}
